package com.example.service;

import org.apache.curator.RetryLoop;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorTransaction;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.curator.framework.recipes.queue.*;
import org.apache.curator.framework.recipes.shared.SharedCount;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;

@Service
public class ZookeeperServiceImpl implements ZookeeperService {

    @Resource
    private CuratorFramework curatorFramework;

    /***
     * 获取客户端
     * @return
     */
    @Override
    public CuratorFramework getCuratorFramework() {
        return this.curatorFramework;
    }

    /***
     * 获取重试
     * @return
     */
    @Override
    public RetryLoop getRetryLoop() {
        return this.curatorFramework.getZookeeperClient().newRetryLoop();
    }

    /***
     * 判断节点是否存在
     * @param path
     * @return
     * @throws Exception
     */
    @Override
    public boolean isExistNode(String path) throws Exception {
        return this.curatorFramework.checkExists().forPath(path) != null;
    }

    /***
     * 创建节点
     * @param mode
     * @param path
     * @throws Exception
     */
    @Override
    public void createNode(CreateMode mode, String path) throws Exception {
        this.curatorFramework.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
    }

    /***
     * 给节点设置数据信息
     * @param path
     * @param nodeData
     * @throws Exception
     */
    @Override
    public void setNodeData(String path, String nodeData) throws Exception {
        this.curatorFramework.setData().forPath(path, nodeData.getBytes(StandardCharsets.UTF_8));
    }

    /***
     * 创建节点并设置数据信息
     * @param mode
     * @param path
     * @param nodeData
     * @return
     * @throws Exception
     */
    @Override
    public String createNodeAndData(CreateMode mode, String path, String nodeData) throws Exception {
        return this.curatorFramework.create().creatingParentsIfNeeded().withMode(mode).forPath(path, nodeData.getBytes(StandardCharsets.UTF_8));
    }

    /***
     * 获取节点数据信息
     * @param path
     * @return
     * @throws Exception
     */
    @Override
    public String getNodeData(String path) throws Exception {
        return new String(this.curatorFramework.getData().forPath(path), StandardCharsets.UTF_8);
    }

    /***
     * 获取节点所有的子节点
     * @param path
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getNodeChild(String path) throws Exception {
        return this.curatorFramework.getChildren().forPath(path);
    }

    /***
     * 获取所有子节点的数据
     * @param path
     * @return
     * @throws Exception
     */
    @Override
    public Map<String, String> getNodeChildMap(String path) throws Exception {
        Map<String, String> map = new HashMap<>();
        List<String> keys = this.curatorFramework.getChildren().forPath(path);
        for (String key : keys) {
            String targetPath = path + PREF + key;
            String nodeData = this.getNodeData(targetPath);
            map.put(key, nodeData);
        }
        return map;
    }

    /***
     * 删除节点
     * @param path
     * @param recursive
     * @throws Exception
     */
    @Override
    public void deleteNode(String path, Boolean recursive) throws Exception {
        if (recursive) {
            this.curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
        } else {
            this.curatorFramework.delete().guaranteed().forPath(path);
        }
    }

    /***
     * 给具体节点设置监听
     * @param path
     * @param nodeCacheListener
     * @throws Exception
     */
    @Override
    public void watchNode(String path, NodeCacheListener nodeCacheListener) throws Exception {
        NodeCache nodeCache = new NodeCache(this.curatorFramework, path);
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start(true);
    }


    /***
     * 监听某个path及下面的节点的路径变化
     * @param path
     * @param pathChildrenCacheListener
     * @throws Exception
     */
    @Override
    public void watchPath(String path, PathChildrenCacheListener pathChildrenCacheListener) throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(this.curatorFramework, path, true);
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    }

    /***
     * 获取主从架构
     * @param path
     * @return
     * @throws Exception
     */
    @Override
    public LeaderLatch getLeaderLatch(String path) throws Exception {
        return new LeaderLatch(this.curatorFramework, path);
    }

    /****
     * zk选主，让主节点执行任务。
     */
    @Override
    public LeaderSelector getLeaderSelector(String leaderPath, ExecutorService executorService, LeaderSelectorListener listener) {
        return new LeaderSelector(this.curatorFramework, leaderPath, executorService, listener);
    }

    /***
     * 获取分布式计数器
     * @param path
     * @param retryNTimes
     * @return
     */
    @Override
    public DistributedAtomicLong getDistributedAtomicLong(String path, RetryNTimes retryNTimes) {
        return new DistributedAtomicLong(this.curatorFramework, path, retryNTimes);
    }

    /***
     * 共享计数器
     * @param path
     * @param val
     * @return
     */
    @Override
    public SharedCount getSharedCount(String path, int val) {
        return new SharedCount(this.curatorFramework, path, val);
    }

    /****
     * 开启事务操作
     * @param path
     * @return
     */
    @Override
    public CuratorTransaction getCuratorTransaction(String path) {
        return this.curatorFramework.inTransaction();
    }

    /***
     * 分布式锁，可重入
     * @param path
     * @return
     */
    @Override
    public InterProcessMutex getInterProcessMutex(String path) {
        return new InterProcessMutex(this.curatorFramework, path);
    }

    /***
     * 读写锁
     * @param path
     * @return
     */
    @Override
    public InterProcessReadWriteLock getInterProcessReadWriteLock(String path) {
        return new InterProcessReadWriteLock(this.curatorFramework, path);
    }

    /***
     * 分布式锁，不可重入
     * @param path
     * @return
     */
    @Override
    public InterProcessSemaphoreMutex getInterProcessSemaphoreMutex(String path) {
        return new InterProcessSemaphoreMutex(this.curatorFramework, path);
    }

    /****
     * 获取栅栏
     * @param path
     * @return
     */
    @Override
    public DistributedBarrier getDistributedBarrier(String path) {
        return new DistributedBarrier(this.curatorFramework, path);
    }

    /***
     * 简单分布式队列
     * @param path
     * @return
     */
    @Override
    public SimpleDistributedQueue getSimpleDistributedQueue(String path) {
        return new SimpleDistributedQueue(this.curatorFramework, path);
    }

    /***
     * 创建刷新特性的队列
     * @param queueConsumer         监听器
     * @param queueSerializer       序列化实例
     * @param path                  监听路径
     * @param minItemsBeforeRefresh 参数表示在刷新（refresh）分布式优先级队列之前，队列中至少需要有多少个元素。当队列中的元素数量少于这个值时，队列不会进行刷新操作。
     * @return
     */
    @Override
    public DistributedPriorityQueue getDistributedPriorityQueue(QueueConsumer queueConsumer, QueueSerializer queueSerializer, String path, int minItemsBeforeRefresh) {
        return QueueBuilder.builder(this.curatorFramework, queueConsumer, queueSerializer, path).buildPriorityQueue(minItemsBeforeRefresh);
    }

    /***
     * 获取分布式队列
     * @param path
     * @return
     */
    @Override
    public DistributedQueue getDistributedQueue(QueueConsumer queueConsumer, QueueSerializer queueSerializer, String path) {
        return QueueBuilder.builder(this.curatorFramework, queueConsumer, queueSerializer, path).buildQueue();
    }

    /****
     * 延时队列
     * @param queueConsumer
     * @param queueSerializer
     * @param path
     * @return
     */
    @Override
    public DistributedDelayQueue getDistributedDelayQueue(QueueConsumer queueConsumer, QueueSerializer queueSerializer, String path) {
        return QueueBuilder.builder(this.curatorFramework, queueConsumer, queueSerializer, path).buildDelayQueue();
    }
}
